package com.bigdata.flink.watermark;

import com.bigdata.flink.util.StreamEnvUtil;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Date;
import java.util.concurrent.TimeUnit;

public class WindowTypeProcessingTimeTest {
  public static void main(String[] args) throws Exception {
//    processTimeWindowTest();
    countWindowTest();
  }

  public static void countWindowTest() throws Exception {
    StreamExecutionEnvironment env = StreamEnvUtil.getEnv();
    DataStreamSource<String> socketSource = StreamEnvUtil.getSocketSource(env);

    SingleOutputStreamOperator process = socketSource.keyBy(f -> f.split("\\s")[0])
        .countWindow(5, 2)
        .process(new ProcessWindowFunction<String, String, String, GlobalWindow>() {
          @Override
          public void process(String string, Context context, Iterable<String> elements, Collector<String> out) throws Exception {
            out.collect(String.format("key=%s, 窗口大小[%s  %s], 元素个数%s, %s", string, "", "", elements.spliterator().estimateSize(), elements));
          }
        });

    process.print();
    env.execute();
  }


  // 窗口：流数据进行数据分块就是窗口。平时用的处理时间、实际上跟随业务应该使用事件时间
  // 窗口分为 滚动窗口、滑动窗口、会话窗口、全局窗口
  // 滚动窗口，窗口大小固定，首位相连没有重叠和间隔
  public static void processTimeWindowTest() throws Exception {
    StreamExecutionEnvironment env = StreamEnvUtil.getEnv();
    DataStreamSource<String> socketSource = StreamEnvUtil.getSocketSource(env);

    SingleOutputStreamOperator process = socketSource.keyBy(f -> f.split("\\s")[0])
        .window(getWindowAssigner1())
        .process(new ProcessWindowFunction<String, String, String, TimeWindow>() {

          @Override
          public void process(String string, Context context, Iterable<String> elements, Collector<String> out) throws Exception {
            long start = context.window().getStart();
            Date date1 = new Date(start);
            long end = context.window().getEnd();
            Date date2 = new Date(end);
            out.collect(String.format("key=%s, 窗口大小[%s  %s], 元素个数%s, %s", string, date1, date2, elements.spliterator().estimateSize(), elements));
          }
        });

    process.print();
    env.execute();
  }

  public static WindowAssigner getWindowAssigner1() {
    return TumblingProcessingTimeWindows.of(Time.of(10, TimeUnit.SECONDS));
  }

  public static WindowAssigner getWindowAssigner2() {
    return SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5));
  }

  public static WindowAssigner getWindowAssigner3() {
    return ProcessingTimeSessionWindows.withGap(Time.seconds(10));
  }
}
